import androidx.lifecycle.DefaultLifecycleObserver
import androidx.lifecycle.LifecycleOwner
import com.wkq.util.socket.CsWebSocketClient
import com.wkq.util.socket.SocketListener
import com.wkq.util.socket.WebSocketClientInterface
import com.wkq.util.socket.WebSocketConfig
import com.wkq.util.socket.WebSocketState
import kotlinx.coroutines.*
import org.java_websocket.handshake.ServerHandshake
import java.net.URI
import java.util.Collections
import java.util.concurrent.atomic.AtomicBoolean

class WebSocketClientImpl(
    private val url: String,
    private val config: WebSocketConfig?,
    private val socketListener: SocketListener?,
    lifecycleOwner: LifecycleOwner? = null
) : WebSocketClientInterface, DefaultLifecycleObserver {

    private var client: CsWebSocketClient? = null
    private val scope = CoroutineScope(Dispatchers.IO + SupervisorJob())
    @Volatile private var state: WebSocketState = WebSocketState.IDLE

    private val isReconnecting = AtomicBoolean(false)
    private var pingJob: Job? = null
    private var reconnectJob: Job? = null
    private var reconnectNum = 0

    private val messageQueue = Collections.synchronizedList(mutableListOf<String>())

    init {
        lifecycleOwner?.lifecycle?.addObserver(this)
    }

    override fun connect() {
        if (state == WebSocketState.CONNECTING || state == WebSocketState.CONNECTED) return

        try {
            if (client == null || client?.isClosed == true) {
                client = object : CsWebSocketClient(URI(url), config?.headers ?: emptyMap()) {
                    override fun onOpen(handshake: ServerHandshake?) {
                        reconnectNum = 0
                        setState(WebSocketState.CONNECTED)
                        startPing()
                        flushMessageQueue()
                        socketListener?.onOpen(handshake)
                    }

                    override fun onMessage(msg: String?) {
                        msg?.let { socketListener?.onMessage(it) }
                    }

                    override fun onClose(code: Int, reason: String?, remote: Boolean) {
                        stopPing()
                        setState(WebSocketState.DISCONNECTED)
                        if (isReconnecting.get()) scheduleReconnect()
                        socketListener?.onClose(code, reason, remote)
                    }

                    override fun onError(ex: Exception?) {
                        super.onError(ex)
                        socketListener?.onError(ex?.message)
                        if (isReconnecting.get()) scheduleReconnect()
                    }
                }
            }

            client?.connectionLostTimeout = (config?.connectTimeout ?: 5000) / 1000
            setState(WebSocketState.CONNECTING)

            if (!(client?.isOpen()?:true)){
                client?.connect()
            }
        } catch (e: Exception) {
            socketListener?.onError(e.message)
            if (isReconnecting.get()) scheduleReconnect()
        }
    }

    override fun disconnect() {
        isReconnecting.set(false)
        reconnectJob?.cancel()

        if (state == WebSocketState.DISCONNECTED || state == WebSocketState.DISCONNECTING) return

        setState(WebSocketState.DISCONNECTING)
        client?.close()
    }
/**
*释放
 */
    override fun release() {
        disconnect()
        scope.cancel()
        client = null
        reconnectJob = null
        pingJob = null
        reconnectNum = 0
        messageQueue.clear()
        isReconnecting.set(false)
        state = WebSocketState.IDLE
    }

    override fun send(message: String): Boolean {
        return if (state == WebSocketState.CONNECTED) {
            try {
                client?.send(message)
                true
            } catch (e: Exception) {
                socketListener?.onError("Send failed: ${e.message}")
                false
            }
        } else {
            messageQueue.add(message)
            false
        }
    }

    private fun flushMessageQueue() {
        synchronized(messageQueue) {
            messageQueue.forEach { msg ->
                try {
                    client?.send(msg)
                } catch (e: Exception) {
                    socketListener?.onError("Send from queue failed: ${e.message}")
                }
            }
            messageQueue.clear()
        }
    }

    private fun setState(newState: WebSocketState) {
        if (state != newState) {
            state = newState
            socketListener?.onStateChange(state)
        }
    }

    private fun startPing() {
        stopPing()
        pingJob = scope.launch {
            while (isActive && state == WebSocketState.CONNECTED) {
                delay(config?.pingTime ?: 10000)
                try {
                    client?.sendPing()
                } catch (e: Exception) {
                    socketListener?.onError("Ping failed: ${e.message}")
                }
            }
        }
    }

    private fun stopPing() {
        pingJob?.cancel()
        pingJob = null
    }
    /**
    * 开启自动重连
     */
     fun enableReconnect() {
        isReconnecting.set(true)
    }
    /**
    *关闭自动重连
     */
     fun disableReconnect() {
        isReconnecting.set(false)
        reconnectJob?.cancel()
    }

    private fun scheduleReconnect() {
        if (!isReconnecting.get() ||
            (config?.maxReconnectNum ?: -1) >= 0 && reconnectNum >= (config?.maxReconnectNum ?: 0)
        ) return

        reconnectJob?.cancel()
        reconnectNum++

        val delayMillis = minOf(
            1000 * (1L shl (reconnectNum - 1)),
            config?.reconnectMaxDelay ?: 10000
        )

        setState(WebSocketState.RECONNECTING)
        reconnectJob = scope.launch {
            delay(delayMillis)
            connect()
        }
    }

    override fun onStart(owner: LifecycleOwner) {
        if (isReconnecting.get()) {
            connect()
        }
    }

    override fun onStop(owner: LifecycleOwner) {
        disconnect()
    }

    override fun onDestroy(owner: LifecycleOwner) {
        release()
    }
}
